import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import com.parabon.client.Job;
import com.parabon.client.RemoteSessionManager;
import com.parabon.client.SerializableTaskSpec;
import com.parabon.client.ServerCommunicationException;
import com.parabon.client.SessionManager;
import com.parabon.client.SimulatorSessionManager;
import com.parabon.client.TaskEvent;
import com.parabon.client.TaskExceptionEvent;
import com.parabon.client.TaskExceptionListener;
import com.parabon.client.TaskProxy;
import com.parabon.client.TaskResultEvent;
import com.parabon.client.TaskResultListener;
import com.parabon.io.InvalidLoginException;



public class ownApp implements TaskResultListener, TaskExceptionListener{
	public static final String JOB_TYPE_ATTRIBUTE         = "JobType";
	public static final String JOB_NAME_ATTRIBUTE         = "JobName";
	public static final String JOB_INPUT_VALUES_ATTRIBUTE = "InputValues";
	public static final String INPUT_VALUE_ATTRIBUTE = "InputValue";
	public static final String EXECUTABLE_ELEMENT_FILENAME = "ownTask.jar";
	private String jobName;
	private SessionManager manager;
	private Job job;
	public static final String JOB_TYPE = "ownApp";
	private Set<Integer> pendingTasks = new HashSet<Integer>();
	public static final String DEFAULT_JOB_NAME = "DefaultownApp";
	private ownApp(String jobName) {
	    this.jobName = jobName;
	    createSession();
	  }
	private void createSession() {
	    System.out.println("Connecting to session manager");
	    try {
	      manager = new SimulatorSessionManager();
	      //manager.reestablish();
	      System.out.println("Connected to remote server");
	    }
	    catch (ServerCommunicationException e) {
	      System.err.println("Unable to establish connection to server");
	      System.err.println(e);
	      System.exit(1);
	    }
	    /*catch (InvalidLoginException e) {
	      System.err.println("Login failed");
	      System.err.println(e);
	      System.exit(1);
	    }*/
	  }
	public static void usage() {
	    System.err.println("ERROR in Usage:(launch|listen|remove)");
	  }
	 public boolean findJob() {
		    		    
		    for (Job currJob : manager.getJobs()) {
		      String currJobType = currJob.getAttributes().get(JOB_TYPE_ATTRIBUTE);
		      String currJobName = currJob.getAttributes().get(JOB_NAME_ATTRIBUTE);
		      if ((currJobType != null) && currJobType.equals(JOB_TYPE) &&
		          (currJobName != null) && currJobName.equals(jobName)) {
		        String[] inputValueStrings =
		          currJob.getAttributes().get(JOB_INPUT_VALUES_ATTRIBUTE).split(" ");
		        for (int i = 0; i < inputValueStrings.length; i++) {
		          int inputValue = Integer.parseInt(inputValueStrings[i]);
		          pendingTasks.add(inputValue);
		        }

		        job = currJob;
		        return true;
		      }
		    }

		    return false;
		  }
	 public void launch(int[] inputValues, boolean addListener) {
		    assert(job == null);
		    
		    System.out.println("Creating " + JOB_TYPE + " with name " + jobName);
		    
		    StringBuilder inputValuesAttribute = new StringBuilder();
		    for (int inputValue : inputValues) {
		      if (inputValuesAttribute.length() != 0) {
		        inputValuesAttribute.append(' ');
		      }
		      inputValuesAttribute.append(inputValue);
		    }
		    
		    Map<String, String> jobAttributes = new HashMap<String, String>();
		    jobAttributes.put(JOB_TYPE_ATTRIBUTE, JOB_TYPE);
		    jobAttributes.put(JOB_NAME_ATTRIBUTE, jobName);
		    jobAttributes.put(JOB_INPUT_VALUES_ATTRIBUTE,
		                      inputValuesAttribute.toString());
		    job = manager.createJob(jobAttributes);
		    System.out.println("job object in launch() is:"+job);
		    File demoDirectory = new File(System.getProperty("demo.home", "."));
		    File taskJarFile = new File(demoDirectory, EXECUTABLE_ELEMENT_FILENAME);
		    String taskJarID = null;
		    try {
		      taskJarID = job.addJavaExecutableElement(taskJarFile);
		    }
		    catch (IOException e) {
		      System.err.println("Can't access task jar file at " + taskJarFile);
		      System.exit(0);
		    }

		    for (int inputValue : inputValues) {
		      pendingTasks.add(inputValue);
		    }
		    
		    if (addListener) {
		      addListener();
		    }
		       
		    //
		    // Start the tasks
		    //
		    SerializableTaskSpec taskSpec = new SerializableTaskSpec();
		    taskSpec.addRequiredExecutableElement(taskJarID);
		    for (int inputValue : inputValues) {
		      Map<String, String> taskAttributes = new HashMap<String, String>();
		      taskAttributes.put(INPUT_VALUE_ATTRIBUTE, Integer.toString(inputValue));

		      taskSpec.setTask(new ownTask(inputValue));
		      
		      System.out.println("Starting task to compute the square of " + inputValue);
		      TaskProxy task = job.addTask(taskSpec, taskAttributes);
		      task.start();
		    }

		    System.out.println("Created " + inputValues.length + " task(s)");
		  }
	 public void addListener() {
		 System.out.println("job object in addListener() is:"+job);
	    job.addListener(this);
	  }
	 public void remove() {
		    if (job != null) {
		      job.remove();
		      job = null;
		    }
		  }
	 public synchronized void waitUntilComplete() {
		    System.out.println("Waiting for results...");
		    while (!pendingTasks.isEmpty()) {
		      try {
		        wait();
		      }
		      catch (InterruptedException e) {}
		    }
    }
	 private synchronized void taskComplete(int inputValue) {
		    System.out.println("into taskComplete()");
		    pendingTasks.remove(inputValue);
		    if (pendingTasks.isEmpty()) {
		      notifyAll();
		    }
		  }
	 private int getInputValue(TaskEvent event) {
		    System.out.println("into getInputValue()");
		    return Integer.parseInt(
		      event.getTaskAttributes().get(INPUT_VALUE_ATTRIBUTE));
		  }
	 public void closeSession() {
		    if (manager != null) {
		      System.out.println("Disconnecting from server");
		      manager.destroy();
		      manager = null;
		      System.out.println("Disconnected");
		    }
		  }
	 public void resultsPosted(TaskResultEvent event) {
		    System.out.println("into resultsPosted()");
		    int inputValue = getInputValue(event);
		    byte[] result = (byte[])event.getResultsObject();
		    System.out.println("got byte[] is:"+result);
		    
		    try{
		    OutputStream out = new FileOutputStream("./finalresults"+inputValue+".txt");
		    out.write(result);
		    out.close();
		    }catch(Exception e){
		    	System.out.println("exception occured:"+e);
		    }
		    taskComplete(inputValue);
		  }

		  public void exceptionThrown(TaskExceptionEvent event) {
		    int inputValue = getInputValue(event);
		    
		    System.out.println(
		      "Exception (" + event.getCode() + ")"
		      + " while working on input value " + inputValue);
		    System.out.println(event.getDescription());
		    System.out.println();

		    taskComplete(inputValue);
		  }

	public static void main(String[] args) {
	    boolean launch = false;
	    boolean listen = false;
	    boolean remove = false;
	    
	    if (args.length == 0) {
	        usage();
	        System.exit(1);
	      }
	    for (String arg : args) {
	        if (arg.equals("launch")) {
	          launch = true;
	        } else if (arg.equals("listen")) {
	          listen = true;
	        } else if (arg.equals("remove")) {
	          remove = true;
	        } else {
	          usage();
	          System.exit(1);
	        }
	      }
	    String jobName = System.getProperty("jobName", DEFAULT_JOB_NAME);
	    ownApp OWNApp = new ownApp(jobName);
	    
	    //Give inputs as an array
	    int[] inputValues = new int[] {1};
	    
	    if (launch) {
	        if (OWNApp.findJob()) {
	          System.err.println(
	            "Job of type " + JOB_TYPE + " with name " + jobName +
	            " already exists.");
	          System.err.println("Use the \"remove\" mode to remove the job.");
	          System.exit(1);
	        }

	       OWNApp.launch(inputValues, listen);
	      } else {
	        if (!OWNApp.findJob()) {
	          System.err.println(
	            "Job of type " + JOB_TYPE + " with name " + jobName +
	            " does not exist.");
	          System.err.println("Use the \"launch\" mode to create a new job.");
	          System.exit(1);
	        }
	      }
	    if (listen) {
	        if (!launch) {
	        	OWNApp.addListener();
	        }
	        OWNApp.waitUntilComplete();
	      }
	    if (remove) {
	    	OWNApp.remove();
	      }
	    OWNApp.closeSession();
	    
	}
}
